package com.jscloud.bigdata.flink.tableapi.overwindow;

import com.jscloud.bigdata.flink.tableapi.groupwindow.TempSensorData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.UNBOUNDED_RANGE;

/**
 * OverWindow使用ProcessTime进行开窗
 * <p>
 * 按processtime排序无界UNBOUNDED_RANGE开窗（FlinkTableOverWinBasePtTime1）
 * <p>
 * 使用overwindow求取温度最高值
 */
public class FlinkTableOverWinProcessTime1 {
        public static void main(String[] args) {
                //1.获取stream的执行环境
                StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
                senv.setParallelism(1);
                //2.创建表执行环境
                StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

                //3.读取数据
                DataStream<TempSensorData> tempSensorData = senv.socketTextStream("bigdata01", 8888)
                        .map(event -> {
                                String[] arr = event.split(",");
                                return TempSensorData.builder()
                                        .sensorID(arr[0])
                                        .tp(Long.parseLong(arr[1]))
                                        .temp(Integer.parseInt(arr[2]))
                                        .build();
                        });

                //4.流转换为动态表
                Table table = tEnv.fromDataStream(tempSensorData,
                        $("sensorID"), $("tp"), $("temp"), $("ptTime").proctime());

                //5.自定义窗口并计算
                Table resultTable = table.window(
                                //定义 over window   partitionBy ==> 指定分区的条件
                                Over.partitionBy($("sensorID"))
                                        //orderBy ==> 指定排序的规则
                                        .orderBy($("ptTime"))
                                        //preceding ==> 指定数据开窗规则, UNBOUNDED_RANGE 没规则
                                        .preceding(UNBOUNDED_RANGE)
                                        .as("w") //指定别名
                        )
                        .select($("sensorID"), $("temp"), $("temp").max().over($("w")).as("maxTemp"));

                //6.执行Flink
                resultTable.execute().print();

        }

}
